package com.dahua.messagesyncsh.kafka.runner;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.dahua.messagesyncsh.common.Constants;
import com.dahua.messagesyncsh.config.redis.RedisUtil;
import com.dahua.messagesyncsh.criteria.CollectionQueryCriteria;
import com.dahua.messagesyncsh.entity.Collection;
import com.dahua.messagesyncsh.entity.SensorData;
import com.dahua.messagesyncsh.pojo.EquipmentInfo;
import com.dahua.messagesyncsh.pojo.Feature;
import com.dahua.messagesyncsh.pojo.LatestSensorData;
import com.dahua.messagesyncsh.pojo.SyncData;
import com.dahua.messagesyncsh.service.CollectionService;
import com.dahua.messagesyncsh.service.EquipmentService;
import com.dahua.messagesyncsh.service.Impl.SensorDataServiceImpl;
import com.dahua.messagesyncsh.service.ProcessorService;
import com.dahua.messagesyncsh.service.SensorDataService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.List;

import static com.dahua.messagesyncsh.common.Constants.DATETIME_FORMATTER_ONE;
import static com.dahua.messagesyncsh.common.Constants.EQUIPMENT_BK2_NULL;

@Slf4j
@Component
public class FeatureHandleRunner implements Runnable {
    @Resource
    private RedisUtil redisUtil;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private EquipmentService equipmentService;

    @Resource
    private SensorDataService sensorDataService;

    @Resource
    private ProcessorService processorService;

    @Resource
    private CollectionService collectionService;

    private final List<ConsumerRecord<String, String>> consumerRecords;

    public FeatureHandleRunner(RedisUtil redisUtil,
                               RedissonClient redisson,
                               EquipmentService equipmentService,
                               SensorDataService sensorDataService,
                               ProcessorService processorService,
                               CollectionService collectionService,
                               List<ConsumerRecord<String, String>> consumerRecords) {
        this.consumerRecords = consumerRecords;
        this.redisUtil = redisUtil;
        this.redissonClient = redisson;
        this.equipmentService = equipmentService;
        this.sensorDataService = sensorDataService;
        this.processorService = processorService;
        this.collectionService = collectionService;
    }

    @Override
    public void run() {
        this.handle(consumerRecords);
    }

    public void handle(List<ConsumerRecord<String, String>> consumerRecords) {
        for (ConsumerRecord<String, String> item : consumerRecords) {
            Feature feature = null;
            try {
                feature = JSONObject.parseObject(item.value(), Feature.class);
            } catch (Exception e) {
                log.error("【Feature】：转换失败 \n {}", e.getMessage());
            }

            if (feature == null) {
                continue;
            }

            /*  如果数据为空则不处理*/
            if (feature.getEquipmentId() == null || feature.getEquipmentId().isEmpty() ||
                    feature.getSensorId() == null || feature.getSensorId().isEmpty() ||
                    feature.getStartTime() == null ||
                    feature.getData() == null) {
                log.error("【Feature】：数据不完整 \n {}", feature);
                continue;
            }

            /*  设备不存在*/
            EquipmentInfo enquire = equipmentService.enquire(feature.getEquipmentId());

            if (enquire == null) {
                continue;
            }

            if (enquire.getBk82().isEmpty()){
                Object hash = redisUtil.getHash(EQUIPMENT_BK2_NULL, enquire.getId().toString());
                if (hash == null){
                    log.error("【Feature】：设备 {} BK2为空 \n", enquire);
                    redisUtil.setHash(EQUIPMENT_BK2_NULL, enquire.getId().toString(), Constants.NULL);
                }
                continue;
            }

            /*  通过对方测点获取我方测点id*/
            Collection collection = null;
            List<Collection> collections = collectionService.enquire(CollectionQueryCriteria.builder().collection(enquire.getBk82()).sensor(feature.getSensorId()).build());
            if (collections.size() != 1) {
                continue;
            }else{
                collection = collections.get(0);
            }
            if (collection == null) {
                log.error("【Feature】：测点对应不存在 \n {}", feature.getSensorId());
                continue;
            }


            /*  一个设备的一个时间段的数据争用*/
            String key = feature.getEquipmentId() + "." + collection.getSensorId() + "." + feature.getStartTime();
            String lockKey = "LOCK-" + key;
            RLock lock = redissonClient.getLock(lockKey);

            try {
                lock.lock();

                SyncData syncData = null;

                /*  数据处理*/
                SensorData sensorData = processorService.buildData(feature, enquire, collection);
                if (sensorData == null) {
                    continue;
                }

                /*  在缓存中保存最新的数据*/
                Object his_object = redisUtil.getHash(String.valueOf(enquire.getId()), collection.getSensor());

                LatestSensorData his_sensorData = null;
                LatestSensorData now_sensorData = null;

                String happenTime = null;
                try{
                    his_sensorData = JSONObject.parseObject(JSON.toJSONString(his_object), LatestSensorData.class);

                    Timestamp timestamp = new Timestamp(feature.getStartTime());
                    Instant instant = timestamp.toInstant();
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern(DATETIME_FORMATTER_ONE).withZone(ZoneId.systemDefault());
                    happenTime = formatter.format(instant);

                }catch (Exception e) {
                    log.error("【Feature】：历史数据转换失败 {}", e.getMessage());
                }

                if (his_sensorData == null){
                    now_sensorData = LatestSensorData.builder()
                            .equipmentId(enquire.getId())
                            .sensorId(collection.getSensor())
                            .sensorDesc(collection.getSensorDesc())
                            .value(sensorData.getRms() == null ? sensorData.getArms() : sensorData.getRms())
                            .data(feature.getData())
                            .unit(collection.getUnit())
                            .timestamp(feature.getStartTime())
                            .happenTime(happenTime)
                            .build();
                }else if (his_sensorData.getTimestamp() != null && (feature.getStartTime() > his_sensorData.getTimestamp())){
                    now_sensorData  = LatestSensorData.builder()
                            .equipmentId(enquire.getId())
                            .sensorId(collection.getSensor())
                            .sensorDesc(collection.getSensorDesc())
                            .value(sensorData.getRms() == null ? sensorData.getArms() : sensorData.getRms())
                            .data(feature.getData())
                            .unit(collection.getUnit())
                            .timestamp(feature.getStartTime())
                            .happenTime(happenTime)
                            .build();
                }else{
                    now_sensorData = his_sensorData;
                }
                /*  缓存最新的数据*/
                redisUtil.setHash(String.valueOf(enquire.getId()), collection.getSensor(), now_sensorData);

                try{
                    sensorDataService.toInsert(key,feature,sensorData, SensorDataServiceImpl.Type.FEATURE);
                }catch (Exception e){
                    log.error("【Feature】：数据更新处理异常 {}", e.getMessage());
                }

            } catch (Exception e) {
                log.error("【Feature】：锁处理异常 {}", e.getMessage());
            } finally {
                lock.unlock();
            }

        }
    }

}
